package org.rency.mq.kafka.producer.service.impl;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.rency.common.utils.domain.BaseResult;
import org.rency.mq.kafka.core.KafkaConstant;
import org.rency.mq.kafka.producer.request.MessageEvent;
import org.rency.mq.kafka.producer.service.MessagePublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

/**
 * ${todo}
 *
 * @author: Created by user_rcy@163.com on 2016/10/14.
 * @version: V1.0.0
 * @since JDK 1.8
 */
public class DefaultMessagePublisher<K, V> implements MessagePublisher<K, V> {

    private static final Logger logger = LoggerFactory.getLogger(DefaultMessagePublisher.class);

    @Resource(name="KafkaTemplate")
    private KafkaTemplate<K, V> kafkaTemplate;

    @Override
    public BaseResult publish(MessageEvent<K, V> request) {
        try {
            return this.send(request,0);
        }catch (Exception e) {
            logger.error("Kafka Producer Message Event Publish Error, Request:{}.",request,e);
            return BaseResult.fail(e.getMessage());
        }
    }

    /**
     * 失败重试
     * @param request
     * @param retryTimes
     * @return
     */
    private BaseResult retry(MessageEvent<K, V> request, int retryTimes){
        logger.info("Kafka Producer Message Event Publish Retry[{}], Request:{}.",retryTimes,request);
        if(KafkaConstant.DEFAULT_AUTO_RETRY_TIMES < retryTimes){
            return BaseResult.fail("Kafka Producer Message Event Publish Error, And Retry Failed.");
        }
        return this.send(request,retryTimes);
    }

    private BaseResult send(MessageEvent<K, V> request, int retryTimes){
        logger.debug("Kafka Producer Message Event Publish, Request:{}.",request);
        long startTime = System.currentTimeMillis();
        try {
            K key = request.getKey();
            V data = request.getContent();
            ListenableFuture<SendResult<K, V>> future = kafkaTemplate.send(request.getTopicName(),request.getPartition(),key,data);
            SendResult<K, V> result = future.get();
            ProducerRecord producerRecord = result.getProducerRecord();
            logger.debug("Kafka Producer Message Event Publish Success, Key:{}, Partition:{}.",producerRecord.key(),producerRecord.partition());
            return BaseResult.success(producerRecord.toString());
        } catch (InterruptedException e) {
            logger.error("Kafka Producer Message Event Publish Error, Request:{}.",request,e);
            return this.retry(request,retryTimes++);
        } catch (ExecutionException e) {
            logger.error("Kafka Producer Message Event Publish Error, Request:{}.",request,e);
            return this.retry(request,retryTimes++);
        } catch (Exception e) {
            throw e;
        }finally {
            logger.debug("Kafka Producer Message Event Publish, Consume {} ms.",(System.currentTimeMillis()-startTime));
        }
    }

}
